package wei.xinya.hbase.config;

import org.apache.commons.collections4.CollectionUtils;

import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 并发执行器
 */
public class ParallelExecutor {

    /**
     * 多线程任务启动服务
     */
    private static final ExecutorService executorService;

    static {
        // JVM可以使用的处理器数目，计算可用线程数
        int avlProc = ManagementFactory.getOperatingSystemMXBean().getAvailableProcessors() + 1;
        // 创建定长线程池，设置最大值
        executorService = Executors.newFixedThreadPool(avlProc);
    }

    /**
     * 执行任务
     *
     * @param task 任务
     */
    public static void runTask(Runnable task) {
        try {
            executorService.execute(task);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 并行执行任务组
     *
     * @param tasks 任务组
     */
    public static void parallelRunTask(List<Runnable> tasks) {
        if (CollectionUtils.isEmpty(tasks)) {
            return;
        }

        try {
            List<Future<?>> futures = tasks.stream().map(executorService::submit).collect(Collectors.toList());

            for (Future<?> future : futures) {
                future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 并行执行任务，获取结果
     *
     * @param tasks 任务
     * @param <T>   结果类型
     * @return List<T>
     */
    public static <T> List<T> runTask(List<Callable<T>> tasks) {
        if (CollectionUtils.isEmpty(tasks)) {
            return null;
        }

        List<T> results = new ArrayList<>();

        try {
            List<Future<T>> futureList = executorService.invokeAll(tasks);
            for (Future<T> future : futureList) {
                results.add(future.get());
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        return results;
    }
}
